package com.example.demo.chacheable;

import com.example.demo.testvo.KeyDeal;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.PessimisticLockingFailureException;
import org.springframework.data.redis.cache.RedisCacheWriter;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @Description @ClassName @Author jsc
 *
 * @date 2020.01.15 09:10
 */
public class MyRedisCacheWriter implements RedisCacheWriter {

  private final RedisConnectionFactory connectionFactory;
  private final Duration sleepTime;

  @Value("${config.showTimePrefix:false}")
  private boolean showTimePrefix;
  /** @param connectionFactory must not be {@literal null}. */
  MyRedisCacheWriter(RedisConnectionFactory connectionFactory) {
    this(connectionFactory, Duration.ZERO);
  }

  /**
   * @param connectionFactory must not be {@literal null}.
   * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use
   *     {@link Duration#ZERO} to disable locking.
   */
  MyRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime) {

    Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
    Assert.notNull(sleepTime, "SleepTime must not be null!");

    this.connectionFactory = connectionFactory;
    this.sleepTime = sleepTime;
  }

  final String regex = "\\[(.*?)]";

  public KeyDeal strTottl(String str) {
    List<String> list = new ArrayList<>();
    KeyDeal kd = new KeyDeal();
    Long ttl = -1L;
    kd.setTtl(ttl);
    Pattern pattern = Pattern.compile(regex);
    Matcher matcher = pattern.matcher(str);
    while (matcher.find()) {
      String strttl = matcher.group(1);
      if (!StringUtils.isEmpty(strttl)) {
        list.add(strttl);
      }
    }
    if (list.size() == 2) {
      // 校验参数 第一个必须是数字
      try {
        ttl = Long.valueOf(list.get(0));
        kd.setTtl(ttl);
        kd.setUtil(list.get(1));
      } catch (Exception e) {
        throw new IllegalArgumentException(str + "--->" + "转换数字错误");
      }
    } else if (list.size() == 1) {
      try {
        ttl = Long.valueOf(list.get(0));
        kd.setTtl(ttl);
      } catch (Exception e) {
        throw new IllegalArgumentException(str + "--->" + "转换数字错误");
      }
    } else {
      return kd;
    }
    return kd;
  }
  /**
   * (non-Javadoc)
   *
   * @see org.springframework.data.redis.cache.RedisCacheWriter#put(java.lang.String, byte[],
   *     byte[], java.time.Duration)
   */
  @Override
  public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {

    Assert.notNull(name, "Name must not be null!");
    Assert.notNull(key, "Key must not be null!");
    Assert.notNull(value, "Value must not be null!");

    execute(
        name,
        connection -> {
          // 当设置了过期时间，则修改取出  数值 单位
          // @Cacheable(value = "getalluser",key = "#root.methodName"+".concat('[400][SS]')")
          // name 对应 value
          // key 对应 value :: key
          // 判断key里面是否设置了过期时间，如果设置了则对key进行缓存，并设置过期时间
          final String skey = new String(key);
          KeyDeal kd = strTottl(skey);
          // 处理key
          // 是否设置默认
          boolean a = shouldExpireWithin(ttl);
          // 是否设置参数
          Long sttl = kd.getTtl();
          // 设置参数  设置默认参数 不设置参数
          if (kd != null && sttl != -1) {
            byte[] nkey = null;
            if (showTimePrefix) {
              nkey =
                  skey.replace("[" + kd.getTtl() + "]", "")
                      .replace("[" + kd.getUtil() + "]", "")
                      .getBytes();
            } else {
              nkey = key;
            }
            connection.set(
                nkey,
                value,
                Expiration.from(sttl, MyTimeUnit.getTimeUnit(kd.getUtil())),
                RedisStringCommands.SetOption.upsert());
          } else if (a) {
            connection.set(
                key,
                value,
                Expiration.from(ttl.toMillis(), TimeUnit.MILLISECONDS),
                RedisStringCommands.SetOption.upsert());
          } else {
            connection.set(key, value);
          }

          return "OK";
        });
  }

  /**
   * (non-Javadoc)
   *
   * @see org.springframework.data.redis.cache.RedisCacheWriter#get(java.lang.String, byte[])
   */
  @Override
  public byte[] get(String name, byte[] key) {

    Assert.notNull(name, "Name must not be null!");
    Assert.notNull(key, "Key must not be null!");

    return execute(name, connection -> connection.get(key));
  }

  /**
   * (non-Javadoc)
   *
   * @see org.springframework.data.redis.cache.RedisCacheWriter#putIfAbsent(java.lang.String,
   *     byte[], byte[], java.time.Duration)
   */
  @Override
  public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl) {

    Assert.notNull(name, "Name must not be null!");
    Assert.notNull(key, "Key must not be null!");
    Assert.notNull(value, "Value must not be null!");

    return execute(
        name,
        connection -> {
          if (isLockingCacheWriter()) {
            doLock(name, connection);
          }

          try {
            if (connection.setNX(key, value)) {

              if (shouldExpireWithin(ttl)) {
                connection.pExpire(key, ttl.toMillis());
              }
              return null;
            }

            return connection.get(key);
          } finally {

            if (isLockingCacheWriter()) {
              doUnlock(name, connection);
            }
          }
        });
  }

  /*
   * (non-Javadoc)
   * @see org.springframework.data.redis.cache.RedisCacheWriter#remove(java.lang.String, byte[])
   */
  @Override
  public void remove(String name, byte[] key) {

    Assert.notNull(name, "Name must not be null!");
    Assert.notNull(key, "Key must not be null!");

    execute(name, connection -> connection.del(key));
  }

  /*
   * (non-Javadoc)
   * @see org.springframework.data.redis.cache.RedisCacheWriter#clean(java.lang.String, byte[])
   */
  @Override
  public void clean(String name, byte[] pattern) {

    Assert.notNull(name, "Name must not be null!");
    Assert.notNull(pattern, "Pattern must not be null!");

    execute(
        name,
        connection -> {
          boolean wasLocked = false;

          try {

            if (isLockingCacheWriter()) {
              doLock(name, connection);
              wasLocked = true;
            }

            byte[][] keys =
                Optional.ofNullable(connection.keys(pattern))
                    .orElse(Collections.emptySet())
                    .toArray(new byte[0][]);

            if (keys.length > 0) {
              connection.del(keys);
            }
          } finally {

            if (wasLocked && isLockingCacheWriter()) {
              doUnlock(name, connection);
            }
          }

          return "OK";
        });
  }

  /**
   * Explicitly set a write lock on a cache.
   *
   * @param name the name of the cache to lock.
   */
  void lock(String name) {
    execute(name, connection -> doLock(name, connection));
  }

  /**
   * Explicitly remove a write lock from a cache.
   *
   * @param name the name of the cache to unlock.
   */
  void unlock(String name) {
    executeLockFree(connection -> doUnlock(name, connection));
  }

  private Boolean doLock(String name, RedisConnection connection) {
    return connection.setNX(createCacheLockKey(name), new byte[0]);
  }

  private Long doUnlock(String name, RedisConnection connection) {
    return connection.del(createCacheLockKey(name));
  }

  boolean doCheckLock(String name, RedisConnection connection) {
    return connection.exists(createCacheLockKey(name));
  }

  /** @return {@literal true} if {@link RedisCacheWriter} uses locks. */
  private boolean isLockingCacheWriter() {
    return !sleepTime.isZero() && !sleepTime.isNegative();
  }

  private <T> T execute(String name, Function<RedisConnection, T> callback) {

    RedisConnection connection = connectionFactory.getConnection();
    try {

      checkAndPotentiallyWaitUntilUnlocked(name, connection);
      return callback.apply(connection);
    } finally {
      connection.close();
    }
  }

  private void executeLockFree(Consumer<RedisConnection> callback) {

    RedisConnection connection = connectionFactory.getConnection();

    try {
      callback.accept(connection);
    } finally {
      connection.close();
    }
  }

  private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection connection) {

    if (!isLockingCacheWriter()) {
      return;
    }

    try {

      while (doCheckLock(name, connection)) {
        Thread.sleep(sleepTime.toMillis());
      }
    } catch (InterruptedException ex) {

      // Re-interrupt current thread, to allow other participants to react.
      Thread.currentThread().interrupt();

      throw new PessimisticLockingFailureException(
          String.format("Interrupted while waiting to unlock cache %s", name), ex);
    }
  }

  private static boolean shouldExpireWithin(@Nullable Duration ttl) {
    return ttl != null && !ttl.isZero() && !ttl.isNegative();
  }

  private static byte[] createCacheLockKey(String name) {
    return (name + "~lock").getBytes(StandardCharsets.UTF_8);
  }
}
